'use strict';

/**
 * Copyright @2016-present, Sumscope, Inc.
 * All rights reserved.
 */

const Redis = require('ioredis');
const _ = require('lodash');
const logger = require('../util/logger');
const {redis} = require('../config');

class RedisClient {
  
  constructor(options) {
    this.subscriber = new Map();
    this.options = options;
    this.client = this.createClient();
    this._defineCommands(this.client);
    this._setupEvents(this.client);
  }
  
  subscribe(topic, callback) {
    callback && this.subscriber.set(topic, callback);
    if (this.subClient === undefined) {
      this._setupSubClient();
    }
    this.subClient.subscribe(topic);
  }
  
  publish(topic, message) {
    this.client.publish(topic, JSON.stringify(message));
  }
  
  ops() {
    return this.client;
  }
  
  origin() {
    return this.client;
  }
  
  originSubClient() {
    if (this.subClient === undefined) {
      this._setupSubClient();
    }
    return this.subClient;
  }
  
  createClient() {
    if (this.options.sentinel.nodes.length > 0) {
      if (!this.options.sentinel.name) {
        throw new Error('Needs master name of redis sentinel.');
      }
      logger.info('Create redis client to sentinel servers ' + this.options.sentinel.nodes);
      return new Redis({
        sentinels: this.options.sentinel.nodes,
        name: this.options.sentinel.name,
        showFriendlyErrorStack: true
      });
    } else {
      let redisOption = {
        port: this.options.port,
        host: this.options.ip,
        showFriendlyErrorStack: true
      };
      if (this.options.password) {
        redisOption.password = this.options.password;
      }
      logger.info('Create redis client to ' + this.options.ip + ':' + this.options.port);
      return new Redis(redisOption);
    }
  }
  
  _defineCommands(client) {
    client.defineCommand('lock', {
      numberOfKeys: 1,
      lua: 'if (redis.call("get", KEYS[1]) == ARGV[1]) then' +
      + '      if (redis.call("setnx", KEYS[1], ARGV[1]) == 1) then'
      + '         redis.call("pexpire", KEYS[1], ARGV[2]);'
      + '         return 1;'
      + '   end;'
      + '   return 0;'
    });
    client.defineCommand('deleteNoTtl', {
      numberOfKeys: 1,
      lua: 'if (redis.call("ttl", KEYS[1]) <= 0) then'
      + '  redis.call("del", KEYS[1]);'
      + '  return 1;'
      + 'end;'
      + 'return 0;'
    })
  }
  
  _setupSubClient() {
    this.subClient = this.createClient(this.options);
    this.subClient.on('message', (topic, message) => {
      const callback = this.subscriber.get(topic);
      if (!_.isNil(callback) && _.isFunction(callback)) {
        callback(message, topic);
      }
    });
    this.subClient.on('reconnecting', () => {
      logger.info('Reconnecting to redis server.');
      for (const topic of this.subscriber.keys()) {
        this.subClient.subscribe(topic);
      }
    });
  }
  
  _setupEvents(client) {
    const name = 'Redis[' + this.options.ip + '] ';
    client.on('error', (err) => {
      logger.error(name + 'got error.', err);
    });
    
    client.on('ready', () => {
      logger.info(name + 'establish connection.');
    });
    
    client.on('end', () => {
      logger.info(name + 'close connection.');
    });
  }
}

module.exports = new RedisClient(redis);
